package com.ccxe.parser;

import java.io.IOException;
import java.util.Iterator;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.MD5Hash;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Partitioner;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;

import com.ccxe.io.Data;
import com.ccxe.io.InjectorKey;
import com.ccxe.util.DATA;
import com.ccxe.util.DateConvertUtil;
import com.ccxe.util.ParserData;
/**
 * 填充小时
 * @author hetao
 *
 */
public class InjectHour {
	public static final Log LOG = LogFactory.getLog(InjectHour.class);

	/************************************************************************************
	 * Driver
	 *******************************************************************************/
	public static int insert(Path selectPath, Path insertPath)
			throws IOException {
		return run(selectPath, insertPath);
	}

	public static void main(String[] args) throws IOException {
		if (args.length != 2) {
			System.err.println("Usage: com.ccxe.driver.InjectHour <in> <out>");
			System.exit(2);
		}
		System.exit(insert(new Path(args[0]), new Path(args[1])));
	}

	private static int run(Path input, Path output) throws IOException {
		
		JobConf conf = new JobConf(InjectHour.class);
		FileSystem fs = FileSystem.get(conf);
		// 如果输出路径存在就删除；
		fs.deleteOnExit(output);
		FileInputFormat.addInputPath(conf, input);
		FileOutputFormat.setOutputPath(conf, output);

		conf.setJobName("InjectHour_" + DateConvertUtil.getDate());
		conf.setMapperClass(InjectHourMapper.class);
		conf.setMapOutputKeyClass(InjectorKey.class);
		conf.setMapOutputValueClass(Data.class);
		conf.setPartitionerClass(InjectHourPartitioner.class);
		conf.setOutputKeyComparatorClass(InjectHourComparator.class);
		conf.setOutputValueGroupingComparator(InjectHourGroupComparator.class);
//		conf.setCombinerClass(InjectHourCombiner.class);
		conf.setReducerClass(InjectHourReducer.class);
		conf.setOutputKeyClass(Data.class);
		conf.setOutputValueClass(NullWritable.class);
		JobClient.runJob(conf);

		return 0;
	}

	/************************************************************************************
	 * InjectHourMapper
	 *******************************************************************************/
	public static class InjectHourMapper extends MapReduceBase implements
			Mapper<LongWritable, Text, InjectorKey, Data> {

		public void map(LongWritable key, Text value,
				OutputCollector<InjectorKey, Data> output, Reporter reporter)
				throws IOException {
			String line = value.toString();
			String[] arrayTmp = ParserData.parser(line);
			// 剔除标题和异常数据
			if (line.indexOf("c_") != -1 || arrayTmp.length < 3)
				return;
			Data data = new Data(arrayTmp);
			// 过滤非交易时间的数据
			int time = (int) ((data.getC_date_time() / 100) % 10000);
			if (key == null
					|| (time >= DATA.OPENINGI_TIME1 && time <= DATA.END_TIME1)
					|| (time >= DATA.OPENINGI_TIME2 && time <= DATA.END_TIME2)) {
				output.collect(new InjectorKey(data.getC_stock_no(),data.getC_date_time()), data);
			}
		}
	}

	/************************************************************************************
	 * InjectHourComparator
	 *******************************************************************************/
	public static class InjectHourComparator extends WritableComparator {
		protected InjectHourComparator() {
			super(InjectorKey.class, true);
		}
		@Override
		public int compare(Object a, Object b) {
//			LOG.info("------------------InjectHourComparator compare()-----------");
			InjectorKey gk1 = (InjectorKey) a;
			InjectorKey gk2 = (InjectorKey) b;
			return gk1.compareTo(gk2);
		}
	}

	/************************************************************************************
	 * InjectHourComparator 按天进行分组。
	 *******************************************************************************/
	public static class InjectHourGroupComparator extends WritableComparator {
		protected InjectHourGroupComparator() {
			super(InjectorKey.class, true);
		}

		@Override
		public int compare(Object a, Object b) {
//			LOG.info("------------------InjectHourComparator compare()-----------");
			InjectorKey gk1 = (InjectorKey) a;
			InjectorKey gk2 = (InjectorKey) b;
			long l1 = Long.parseLong(gk1.getC_stock_no() + ""
					+ gk1.getC_date_time() / 1000000);
			long l2 = Long.parseLong(gk2.getC_stock_no() + ""
					+ gk2.getC_date_time() / 1000000);
			return (l1 == l2) ? 0 : (l1 < l2 ? -1 : 1);
		}
	}

	/************************************************************************************
	 * InjectHourComparator 按天进行分区
	 *******************************************************************************/
	public static class InjectHourPartitioner implements
			Partitioner<InjectorKey, Data> {
		@Override
		public int getPartition(InjectorKey key, Data value, int numPartitions) {
//			LOG.info("------------------InjectHourComparator getPartition()-----------");
			return Math.abs(MD5Hash.digest(
					key.getC_stock_no() + "" + key.getC_date_time() / 1000000)
					.quarterDigest())
					% numPartitions;
		}

		@Override
		public void configure(JobConf jobConfw) {
		}
	}

	/************************************************************************************
	 * InjectHourCombiner
	 *******************************************************************************/
	public static class InjectHourCombiner extends MapReduceBase implements
			Reducer<InjectorKey, Data, InjectorKey, Data> {

		@Override
		public void reduce(InjectorKey key, Iterator<Data> values,
				OutputCollector<InjectorKey, Data> output, Reporter reporter)
				throws IOException {
//			LOG.info("------------------InjectHourCombiner reduce()-----------");
			while (values.hasNext()) {
				output.collect(key, values.next());
			}
		}
	}

	/************************************************************************************
	 * InjectHourReducer
	 *******************************************************************************/
	public static class InjectHourReducer extends MapReduceBase implements
			Reducer<InjectorKey, Data, Data, NullWritable> {

		private Data tmpData;
		private int count = 0;
		
		@Override
		public void reduce(InjectorKey key, Iterator<Data> values,
				OutputCollector<Data, NullWritable> output, Reporter reporter)
				throws IOException {
			LOG.info("---------------InjectHourReducer run--------------------");
			// 2010-01-02 12:13:14
			while (values.hasNext()) {
				count++;
				Data data = values.next();
				int flag = 0;
				if (tmpData == null) {
					tmpData = initData(data.getC_stock_no(),
							data.getC_date_time());
					if(!(data.getC_date_time()%1000000/10000 == 15)) {
						output.collect(tmpData,NullWritable.get());
//						LOG.info("User-default Data : "+tmpData);
					}
				}
				/*
				 * 必须保证容器values中的数据为同一ID和同一天； 
				 * 如果两条数据的小时差flag大于1则表示当中缺少小时数据。
				 * 而股市的交易时间将在output时考虑。
				 */
				// 获取2条数据之间的小时差
				flag = (int) ((tmpData.getC_date_time() / 10000) - (data
						.getC_date_time() / 10000));
				if (flag > 1) {
					inject(tmpData, flag, output,key);
				}
				output.collect(data, NullWritable.get());
				try {
					tmpData = (Data) data.clone();
				} catch (CloneNotSupportedException e) {
					e.printStackTrace();
				}
			}
			int lastFlag = (int)((tmpData.getC_date_time() % 1000000) / 10000 - DATA.OPENINGI_TIME1/100);
			if(lastFlag > 0) {
				inject(tmpData,lastFlag+1,output,key);
			}
			LOG.info("--------------InjectHourReducer hashNext() end-----------------------------");
			LOG.info(" [ "+tmpData.getC_stock_no()+" ]号股票在"+tmpData.getC_date_time()/1000000+"内，数据文件中有 【  "+count+"  】条数据");
			tmpData = null;
		}

		private void inject(Data prev, int flag,
				OutputCollector<Data, NullWritable> output,InjectorKey key) throws IOException {
			for (int i = 1; i < flag ; i++) {
				Data tmp = new Data(prev.getC_stock_no());
				long time = ((prev.getC_date_time()/10000- i ) * 10000)+5959;
				if(time%1000000 == 115959) {
					time = ((prev.getC_date_time()/10000- i ) * 10000)+3059;
				}
				tmp.setC_date_time(time);
				if(isValid(tmp.getC_date_time())) {
					output.collect(tmp, NullWritable.get());
//					LOG.info("User-default Data : "+tmp);
				}
			}
		}

		//如果在交易时间内返回true;
		private boolean isValid(long time) {
			time = time %1000000/100;
			return (time >= DATA.OPENINGI_TIME1 && time <= DATA.END_TIME1)
					|| (time >= DATA.OPENINGI_TIME2 && time <= DATA.END_TIME2);
		}

		private Data initData(int no, long time) {
			// 1234-12-12 12:34:56
//			long hour = time % 1000000 / 10000;
			long date = time / 1000000 * 1000000;
			// if (hour >= DATA.END_TIME2) {
			return new Data(no, date + 150059);
			/* }
			 if (hour >= DATA.END_TIME1 && hour < DATA.OPENINGI_TIME2) {
			 return tmpData = new Data(no, date + 113059, 0.0F);
			 } else {
			 return tmpData = new Data(no, date + 5959, 0.0F);
			 }*/
		}
	}

}
